package net.sf.calibur.io;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class Pointer implements Comparable<Pointer> {
	int chunkSize = 0;
	int chunkId = 0;
	int offset = 0;

	Pointer(int chunkSize) {
		this.chunkSize = chunkSize;
		reset();
	}

	void reset() {
		chunkId = 0;
		offset = 0;
	}

	int length() {
		return chunkId * chunkSize + offset;
	}
	
	void setLength(int length) {
		chunkId = length / chunkSize;
		offset = length % chunkSize;
	}

	@Override
	public int compareTo(Pointer another) {
		return length() - another.length();
	}
}

public class ReadWriteBuffer extends OutputStream {
	private final static int DEFAULT_CHUNK_SIZE = 1024;
	private final static int EOF = -1;

	private int chunkSize;
	private List<byte[]> chunks = new ArrayList<byte[]>();
	private Pointer readPointer, writePointer;
	private Lock lock = new ReentrantLock();

	public ReadWriteBuffer() {
		this(DEFAULT_CHUNK_SIZE);
	}

	public ReadWriteBuffer(int chunkSize) {
		if (chunkSize <= 0) {
			throw new IllegalArgumentException("Illegal chunk size "
					+ chunkSize);
		}

		this.chunkSize = chunkSize;
		byte[] firstChunk = new byte[chunkSize];
		chunks.add(firstChunk);
		readPointer = new Pointer(chunkSize);
		writePointer = new Pointer(chunkSize);
	}

	@Override
	public void write(int b) throws IOException {
		byte[] chunk = null;
		lock.lock();
		try {
			if (writePointer.offset == chunkSize) {
				++writePointer.chunkId;
				writePointer.offset = 0;

				if (writePointer.chunkId == chunks.size()) {
					chunk = addChunk();
				}
			}

			chunk = getChunk(writePointer.chunkId);
			chunk[writePointer.offset] = (byte) b;
			++writePointer.offset;
		} finally {
			lock.unlock();
		}
	}
	
	private byte[] addChunk() {
		lock.lock();
		try {
			byte[] chunk = new byte[chunkSize];
			chunks.add(chunk);
			return chunk;
		} finally {
			lock.unlock();
		}
	}

	public int length() {
		return writePointer.length();
	}
	
	public void setLength(int length) {
		writePointer.setLength(length);
	}
	
	public void expandCapacity(int capacity) {
		lock.lock();
		try {
			while (capacity - chunks.size() * chunkSize > 0) {
				addChunk();
			}
		} finally {
			lock.unlock();
		}
	}
	
	public int getCapacity() {
		return chunks.size() * chunkSize;
	}
	
	public int getRemaining() {
		return writePointer.compareTo(readPointer);
	}

	public void reset() {
		lock.lock();
		try {
			writePointer.reset();
			readPointer.reset();
			for (byte[] chunk : chunks) {
				Arrays.fill(chunk, (byte) 0);
			}
		} finally {
			lock.unlock();
		}
	}

	public int getChunkSize() {
		return chunkSize;
	}

	public int getChunksNum() {
		return chunks.size();
	}
	
	public byte[] getChunk(int i) {
		return chunks.get(i);
	}

	public InputStream asInputStream() {
		return new InputStream() {
			@Override
			public int read() throws IOException {
				lock.lock();
				try {
					if (getRemaining() <= 0) {
						return EOF;
					}

					byte[] chunk = chunks.get(readPointer.chunkId);
					int b = chunk[readPointer.offset];
					++readPointer.offset;
					if (readPointer.offset == chunkSize) {
						++readPointer.chunkId;
						readPointer.offset = 0;
					}

					return b & 0x000000ff;
				} finally {
					lock.unlock();
				}
			}

			@Override
			public int available() throws IOException {
				return writePointer.compareTo(readPointer);
			}

			@Override
			public void close() throws IOException {
				// Nothing to do.
			}

			@Override
			public synchronized void mark(int readlimit) {
				throw new UnsupportedOperationException();
			}

			@Override
			public boolean markSupported() {
				throw new UnsupportedOperationException();
			}

			@Override
			public int read(byte[] b, int off, int len) throws IOException {
				return super.read(b, off, len);
			}

			@Override
			public int read(byte[] b) throws IOException {
				return super.read(b);
			}

			@Override
			public synchronized void reset() throws IOException {
				readPointer.reset();
			}

			@Override
			public long skip(long n) throws IOException {
				int i = 0;
				lock.lock();
				try {
					for (i = 0; i < n; ++i) {
						read();
					}
					return n;
				} catch (IndexOutOfBoundsException e) {
					return i;
				} finally {
					lock.unlock();
				}
			}
		};
	}

	// UnitTest codes.
	public static void main(String[] args) throws Exception {
		ReadWriteBuffer buffer = new ReadWriteBuffer(4096);

		doTest(buffer);
		buffer.reset();
		doTest(buffer);

	}

	private static void doTest(ReadWriteBuffer buffer) throws Exception {
		long bm = System.currentTimeMillis();
		DataOutputStream dos = new DataOutputStream(buffer);
		for (long i = 0; i < 100000; ++i) {
			dos.writeLong(i);
		}

		dos.writeUTF("世界你好！");
		

		DataInputStream dis = new DataInputStream(buffer.asInputStream());
		long v = 0;
		for (long i = 0; i < 100000; ++i) {
			try {
				v = dis.readLong();
			} catch (Exception e) {
				System.err.println("error i=" + i);
				e.printStackTrace();
				throw new RuntimeException(e);
			}

			if (v != i) {
				throw new RuntimeException("ERROR at " + i + " v=" + v);
			}
		}
		System.out.println();
		System.out.println(dis.readUTF());
		System.out.println("spent=" + (System.currentTimeMillis() - bm));
		System.out.printf("chunkSize=%d chunksNum=%d Size=%d\n",
				buffer.getChunkSize(), buffer.getChunksNum(), buffer.length());
	}
}
